package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"time"

	"github.com/louloulin/dataflare/pkg/workflow/connector"
)

func main() {
	fmt.Println("=== DataFlare File Connectors Verification ===")
	
	// Create verification directory
	dataDir := "verification/data"
	err := os.MkdirAll(dataDir, 0755)
	if err != nil {
		log.Fatalf("Failed to create data directory: %v", err)
	}

	// Verify file connectors
	verifyFileConnector(dataDir)
	
	// Verify CSV connectors
	verifyCSVConnector(dataDir)
	
	fmt.Println("\n=== File Connectors Verification Complete ===")
}

// Verify file connector
func verifyFileConnector(dataDir string) {
	fmt.Println("\n=== Verifying File Connector ===")
	
	// Create test data
	testData := []map[string]interface{}{
		{"id": 1, "name": "Alice", "email": "alice@example.com"},
		{"id": 2, "name": "Bob", "email": "bob@example.com"},
		{"id": 3, "name": "Charlie", "email": "charlie@example.com"},
	}
	
	// Create test data file
	inputFile := filepath.Join(dataDir, "input.json")
	outputFile := filepath.Join(dataDir, "output.json")
	
	// Write test data to input file
	file, err := os.Create(inputFile)
	if err != nil {
		log.Fatalf("Failed to create input file: %v", err)
	}
	encoder := json.NewEncoder(file)
	for _, data := range testData {
		err := encoder.Encode(data)
		if err != nil {
			log.Fatalf("Failed to encode data: %v", err)
		}
	}
	file.Close()
	
	fmt.Printf("Created test data file: %s\n", inputFile)
	
	// Get connector factory
	factory := connector.DefaultRegistry.GetFactory()
	
	// Create source connector
	fmt.Println("Creating file source connector...")
	source, err := factory.CreateSourceConnector("file")
	if err != nil {
		log.Fatalf("Failed to create file source connector: %v", err)
	}
	
	// Initialize source connector
	err = source.Initialize(map[string]interface{}{
		"path":       inputFile,
		"format":     "json",
		"batch_size": 10,
	})
	if err != nil {
		log.Fatalf("Failed to initialize file source connector: %v", err)
	}
	
	// Create sink connector
	fmt.Println("Creating file sink connector...")
	sink, err := factory.CreateSinkConnector("file")
	if err != nil {
		log.Fatalf("Failed to create file sink connector: %v", err)
	}
	
	// Initialize sink connector
	err = sink.Initialize(map[string]interface{}{
		"path":   outputFile,
		"format": "json",
	})
	if err != nil {
		log.Fatalf("Failed to initialize file sink connector: %v", err)
	}
	
	// Start connectors
	ctx := context.Background()
	fmt.Println("Starting connectors...")
	err = source.Start(ctx)
	if err != nil {
		log.Fatalf("Failed to start source connector: %v", err)
	}
	defer source.Stop(ctx)
	
	err = sink.Start(ctx)
	if err != nil {
		log.Fatalf("Failed to start sink connector: %v", err)
	}
	defer sink.Stop(ctx)
	
	// Process records
	fmt.Println("Processing records...")
	recordCount := 0
	for {
		// Read batch
		batch, err := source.Read(ctx)
		if err != nil {
			log.Fatalf("Failed to read records: %v", err)
		}
		
		// Check if batch is empty (end of data)
		if batch == nil || len(batch.Records) == 0 {
			break
		}
		
		// Process records
		for _, record := range batch.Records {
			// Parse record
			var data map[string]interface{}
			err := json.Unmarshal(record.Value, &data)
			if err != nil {
				log.Fatalf("Failed to parse record: %v", err)
			}
			
			// Add processed flag
			data["processed"] = true
			data["timestamp"] = time.Now().Format(time.RFC3339)
			
			// Update record
			updatedValue, err := json.Marshal(data)
			if err != nil {
				log.Fatalf("Failed to marshal record: %v", err)
			}
			record.Value = updatedValue
		}
		
		// Write batch
		err = sink.Write(ctx, batch)
		if err != nil {
			log.Fatalf("Failed to write records: %v", err)
		}
		
		recordCount += len(batch.Records)
	}
	
	// Flush sink
	err = sink.Flush(ctx)
	if err != nil {
		log.Fatalf("Failed to flush sink: %v", err)
	}
	
	fmt.Printf("Processed %d records\n", recordCount)
	fmt.Printf("Output written to: %s\n", outputFile)
	
	// Verify output file exists
	_, err = os.Stat(outputFile)
	if os.IsNotExist(err) {
		log.Fatalf("Output file does not exist: %s", outputFile)
	}
	
	fmt.Println("File connector verification completed successfully!")
}

// Verify CSV connector
func verifyCSVConnector(dataDir string) {
	fmt.Println("\n=== Verifying CSV Connector ===")
	
	// Create test data
	inputFile := filepath.Join(dataDir, "input.csv")
	outputFile := filepath.Join(dataDir, "output.csv")
	
	// Write test data to input file
	file, err := os.Create(inputFile)
	if err != nil {
		log.Fatalf("Failed to create input file: %v", err)
	}
	
	// Write CSV header
	file.WriteString("id,name,email\n")
	
	// Write CSV data
	file.WriteString("1,Alice,alice@example.com\n")
	file.WriteString("2,Bob,bob@example.com\n")
	file.WriteString("3,Charlie,charlie@example.com\n")
	
	file.Close()
	
	fmt.Printf("Created test data file: %s\n", inputFile)
	
	// Get connector factory
	factory := connector.DefaultRegistry.GetFactory()
	
	// Create source connector
	fmt.Println("Creating CSV source connector...")
	source, err := factory.CreateSourceConnector("csv")
	if err != nil {
		log.Fatalf("Failed to create CSV source connector: %v", err)
	}
	
	// Initialize source connector
	err = source.Initialize(map[string]interface{}{
		"path":      inputFile,
		"header":    true,
		"delimiter": ",",
	})
	if err != nil {
		log.Fatalf("Failed to initialize CSV source connector: %v", err)
	}
	
	// Create sink connector
	fmt.Println("Creating CSV sink connector...")
	sink, err := factory.CreateSinkConnector("csv")
	if err != nil {
		log.Fatalf("Failed to create CSV sink connector: %v", err)
	}
	
	// Initialize sink connector
	err = sink.Initialize(map[string]interface{}{
		"path":      outputFile,
		"header":    true,
		"delimiter": ",",
	})
	if err != nil {
		log.Fatalf("Failed to initialize CSV sink connector: %v", err)
	}
	
	// Start connectors
	ctx := context.Background()
	fmt.Println("Starting connectors...")
	err = source.Start(ctx)
	if err != nil {
		log.Fatalf("Failed to start source connector: %v", err)
	}
	defer source.Stop(ctx)
	
	err = sink.Start(ctx)
	if err != nil {
		log.Fatalf("Failed to start sink connector: %v", err)
	}
	defer sink.Stop(ctx)
	
	// Process records
	fmt.Println("Processing records...")
	recordCount := 0
	for {
		// Read batch
		batch, err := source.Read(ctx)
		if err != nil {
			log.Fatalf("Failed to read records: %v", err)
		}
		
		// Check if batch is empty (end of data)
		if batch == nil || len(batch.Records) == 0 {
			break
		}
		
		// Process records
		for _, record := range batch.Records {
			// Parse record
			var data map[string]interface{}
			err := json.Unmarshal(record.Value, &data)
			if err != nil {
				log.Fatalf("Failed to parse record: %v", err)
			}
			
			// Add processed flag
			data["processed"] = "true"
			data["timestamp"] = time.Now().Format(time.RFC3339)
			
			// Update record
			updatedValue, err := json.Marshal(data)
			if err != nil {
				log.Fatalf("Failed to marshal record: %v", err)
			}
			record.Value = updatedValue
		}
		
		// Write batch
		err = sink.Write(ctx, batch)
		if err != nil {
			log.Fatalf("Failed to write records: %v", err)
		}
		
		recordCount += len(batch.Records)
	}
	
	// Flush sink
	err = sink.Flush(ctx)
	if err != nil {
		log.Fatalf("Failed to flush sink: %v", err)
	}
	
	fmt.Printf("Processed %d records\n", recordCount)
	fmt.Printf("Output written to: %s\n", outputFile)
	
	// Verify output file exists
	_, err = os.Stat(outputFile)
	if os.IsNotExist(err) {
		log.Fatalf("Output file does not exist: %s", outputFile)
	}
	
	fmt.Println("CSV connector verification completed successfully!")
}
